package com.ng.common.component.mqtt.callback;

import lombok.extern.slf4j.Slf4j;

import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import com.ng.biz.service.IMqttClient;
import com.ng.biz.service.impl.MqttClientImpl;
import com.ng.common.component.mqtt.LDMPMqttClient;
import com.ng.common.component.mqtt.handler.MessageArrivedHandler;
import com.ng.common.util.SpringContextHolder;

@Slf4j
public class LDMPMqttCallback implements MqttCallbackExtended {

	private LDMPMqttClient mqttClient;
	
	private final ExecutorService executorService = Executors.newCachedThreadPool(); // 创建线程池
	
	private MessageArrivedHandler messageArrivedHandler;
	private IMqttClient mqttClientImpl;
	
	public LDMPMqttCallback(LDMPMqttClient mqttCollect) {
		this.mqttClient = mqttCollect;
		messageArrivedHandler=new MessageArrivedHandler();
		mqttClientImpl=SpringContextHolder.getBean(MqttClientImpl.class);
	}

	/**
	 * 丢失连接，可在这里做重连 只会调用一次
	 * @param throwable
	 */
	@Override
	public void connectionLost(Throwable throwable) {
		log.warn("{} Connection lost: {}",mqttClient.getClientId() ,throwable.getMessage());
		if (!mqttClient.isReconnecting()) {
			try {
				mqttClient.setReconnecting(true);
				mqttClient.connect();
				mqttClient.subscribeTopics();
			} catch (MqttException e) {
				log.error("{} Error while reconnecting in callback: {}",mqttClient.getClientId(), e.getMessage());
			} finally {
				mqttClient.setReconnecting(false);
			}
		}
	}

	@Override
	public void connectComplete(boolean reconnect, String serverURI) {
		log.info("{} 连接成功，连接方式：{}",mqttClient.getClientId(), reconnect ? "重连" : "直连");
	}

	/**
	 * publish后，配送完成后回调的方法
	 * @param iMqttDeliveryToken
	 */
	@Override
	public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
		//log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
	}

	/**
	 * @param topic
	 * @param mqttMessage
	 * @throws Exception subscribe后得到的消息会执行到这里面
	 */
	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		executorService.submit(() -> {
			if(mqttClient!=null) {
				String key="";
				Iterator<String> it=mqttClient.getTopicMapingTopic().keySet().iterator();
				while(it.hasNext()) {
					key=it.next();
					if(topic.startsWith(key)) {
						messageArrivedHandler.handleWithFlow(mqttClient.getTaskId(),topic,mqttClient.getTopicMapingTopic().get(key), new String(message.getPayload()));
						break;
					}
				}
			}
		});
		mqttClientImpl.incrementAndGet();
	}

}
